package com.eshin.container;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.function.Consumer;

import org.aopalliance.aop.Advice;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.BackOffExecution;

import com.eshin.connectionfactory.MyConnectionFactory;
import com.rabbitmq.client.Channel;

public class MyMessageListenerContainer extends SimpleMessageListenerContainer {

	private List<MySimpleMessageListenerContainer> listenerContainers = new ArrayList<>();
	private MyConnectionFactory connectionFactory;
	public MyMessageListenerContainer(MyConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}
	
	public void init(){
		System.out.println("ddd");
		connectionFactory.getConnectionFactoryList().forEach(new Consumer<CachingConnectionFactory>() {

			@Override
			public void accept(CachingConnectionFactory connectionFactory) {
				MySimpleMessageListenerContainer container = new MySimpleMessageListenerContainer(connectionFactory);
				listenerContainers.add(container);
			}
		});
	}
	@Override
	public void doStart() throws Exception {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
			   try {
				t.doStart();
			} catch (Exception e) {
				e.printStackTrace();
			}
			}
		});
		
		
	}
	
	@Override
	public void setAdviceChain(Advice[] adviceChain) {
		// TODO Auto-generated method stub
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setAdviceChain(adviceChain);
			}
			
		});
	}
	
	@Override
	public void setRecoveryInterval(long recoveryInterval) {
		// TODO Auto-generated method stub
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setRecoveryInterval(recoveryInterval);
			}
			
		});
	}
	
	@Override
	public void setRecoveryBackOff(BackOff recoveryBackOff) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setRecoveryBackOff(recoveryBackOff);
			}
			
		});
	}
	
	@Override
	public void setConcurrentConsumers(int concurrentConsumers) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setConcurrentConsumers(concurrentConsumers);
			}
			
		});
	}
	
//	@Override
//	public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				t.setMaxConcurrentConsumers(maxConcurrentConsumers);
//			}
//			
//		});
//	}
	
//	@Override
//	public void setReceiveTimeout(long receiveTimeout) {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				t.setReceiveTimeout(receiveTimeout);
//			}
//			
//		});
//	}
	
	@Override
	public void setTaskExecutor(Executor taskExecutor) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setTaskExecutor(taskExecutor);
			}
			
		});
	}
	
//	@Override
//	public void setPrefetchCount(int prefetchCount) {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				t.setPrefetchCount(prefetchCount);
//			}
//			
//		});
//	}
	
//	@Override
//	public void setTransactionManager(PlatformTransactionManager transactionManager) {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				t.setTransactionManager(transactionManager);
//			}
//			
//		});
//	}
	
//	@Override
//	public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				t.setTransactionAttribute(transactionAttribute);
//			}
//			
//		});
//	}
	
	@Override
	public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setMessagePropertiesConverter(messagePropertiesConverter);
			}
			
		});
	}
	
	@Override
	public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setDefaultRequeueRejected(defaultRequeueRejected);
			}
			
		});
	}
	
	@Override
	public void setConsumerArguments(Map<String, Object> args) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setConsumerArguments(args);
			}
			
		});
	}
	
/*	@Override
	public void setRabbitAdmin(RabbitAdmin rabbitAdmin) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setRabbitAdmin(rabbitAdmin);
			}
			
		});
	}*/
	
	@Override
	public void setQueueNames(String... queueName) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setQueueNames(queueName);
			}
			
		});
	}
	
	@Override
	public void setQueues(Queue... queues) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setQueues(queues);
			}
			
		});
	}
	
	@Override
	public void addQueueNames(String... queueName) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.addQueueNames(queueName);
			}
			
		});
	}
	
	@Override
	public void addQueues(Queue... queue) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.addQueues(queue);
			}
			
		});
	}
	
	@Override
	public void setConsumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setConsumerTagStrategy(consumerTagStrategy);
			}
			
		});
	}
	
	@Override
	protected void doInitialize() throws Exception {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				try {
					t.doInitialize();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
		});
	}
	
	@Override
	protected void doStop() {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.doStop();
			}
			
		});
	}
	
	@Override
	protected void doShutdown() {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.doShutdown();
			}
			
		});
	}
	
	@Override
	protected int initializeConsumers() {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.initializeConsumers();
			}
			
		});
		return listenerContainers.get(0).getActiveConsumerCount();
	}
	
	@Override
	protected void addAndStartConsumers(int delta) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.addAndStartConsumers(delta);
			}
			
		});
	}
	@Override
	protected void invokeListener(Channel channel, Message message) throws Exception {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				try {
					t.invokeListener(channel, message);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
		});
	}
	
//	@Override
//	protected void handleStartupFailure(BackOffExecution backOffExecution) throws Exception {
//		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {
//
//			@Override
//			public void accept(MySimpleMessageListenerContainer t) {
//				try {
//					t.handleStartupFailure(backOffExecution);
//				} catch (Exception e) {
//					e.printStackTrace();
//				}
//			}
//			
//		});
//	}
	
	@Override
	public void setMessageListener(Object messageListener) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setMessageListener(messageListener);
			}
			
		});
	}
	
	@Override
	public void setMessageConverter(MessageConverter messageConverter) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setMessageConverter(messageConverter);
			}
			
		});
	}
	
	@Override
	public void setAfterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setAfterReceivePostProcessors(afterReceivePostProcessors);
			}
			
		});
	}
	
	@Override
	public void setDeBatchingEnabled(boolean deBatchingEnabled) {
		listenerContainers.forEach(new Consumer<MySimpleMessageListenerContainer>() {

			@Override
			public void accept(MySimpleMessageListenerContainer t) {
				t.setDeBatchingEnabled(deBatchingEnabled);
			}
			
		});
	}
}
